package com.example.myapplication;

import android.annotation.SuppressLint;
import android.util.Log;


import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.Scheduler;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Action;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Consumer;
import io.reactivex.internal.schedulers.IoScheduler;
import io.reactivex.schedulers.Schedulers;

/**
 * @author:weng
 * @data:2022/11/4
 * @description:rxjava操作符测试
 */

public class RxJavaFactory {

    private Disposable disposable;

    //onCreate测试
    @SuppressLint("CheckResult")
    public void onCreate() {
        //通过Observable.create() 创建一个被观察者 Observable 对象
        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            /** 在subscribe（）里定义需要发送的事件 */
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                // 通过 ObservableEmitter类对象产生 & 发送事件
                // ObservableEmitter类负责定义事件发射器 & 向观察者发送事件
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();
            }
        });
//        返回的observable 其实就是创建的 new ObservableOnSubscribe
        observable.doOnDispose(() -> Log.e("tag", "doOnDispose")).subscribe(new Observer<Integer>() {//订阅一个观察者对象

            // 默认最先调用复写的 onSubscribe（）
            @Override
            public void onSubscribe(Disposable d) {
                disposable = d;
            }


            @Override
            public void onNext(Integer value) {
                Log.e("tag", "value--->" + value);
                if (value == 2) {
                    disposable.dispose();
                }
            }

            @Override
            public void onError(Throwable e) {
            }

            @Override
            public void onComplete() {
                Log.e("tag", "onComplete");


            }

        });
    }

    @SuppressLint("CheckResult")
    public void just(){

         Observable<String> observable1 = Observable.create(new ObservableOnSubscribe<String>() {
             @Override
             public void subscribe(ObservableEmitter<String> emitter) throws Exception {

                 emitter.onNext("1");
             }
         }).delay(2, TimeUnit.SECONDS)
                 .subscribeOn(Schedulers.io());
        Observable<String> observable2 = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {

                emitter.onNext("2");
            }
        }).subscribeOn(Schedulers.io());

        Observable.zip(observable1, observable2, new BiFunction<String, String, Object>() {
                    @Override
                    public Object apply(String s, String s2) throws Exception {

                        Log.e("tag","s--->"+s);
                        Log.e("tag","s2--->"+s2);

                        return null;
                    }
                }).observeOn(AndroidSchedulers.mainThread())
                .subscribe();
    }


}
